﻿
Imports std
Imports centrosphere
Imports command = centrosphere.job.storage.command

Partial Public Class storageJobRedundanceDistributor
    Private Overloads Function hashcode(ByVal i As String) As UInt32
        assert(Not isemptystring(i))
        Return _ijob.hashcode(i, targetQueues().size())
    End Function

    Private Function valid_sourceIndex(Of T)(ByVal i As T,
                                             ByVal sourceIndex As UInt32) As Boolean
        Return Not foreach_redundance_queue(i, Function(ByRef j) j <> sourceIndex)
    End Function

    Private Function valid_sourceIndexes(ByVal key As String,
                                         ByVal sourceIndexes As [set](Of UInt32)) As Boolean
        assert(Not sourceIndexes Is Nothing AndAlso sourceIndexes.size() > 0)
        Dim it As [set](Of UInt32).iterator = Nothing
        it = sourceIndexes.begin()
        While it <> sourceIndexes.end()
            If Not valid_sourceIndex(key, +it) Then
                Return False
            End If
            it += 1
        End While
        Return True
    End Function

    Private Sub sync_read_modify(ByVal key As String,
                                 ByVal sourceIndexes As [set](Of UInt32))
        assert(valid_sourceIndexes(key, sourceIndexes))
        Dim j As storageJob = Nothing
        j = New storageJob()
        j.command = command.read
        j.key = key
        assert(j.valid())
        Dim times As Int64 = 0
        times = redundanceCount()
        assert_begin(New eventComb(ijobStep(j, targetQueue(+sourceIndexes.begin()), AddressOf goto_next),
                                   Function() As Boolean
                                       If j.result Then
                                           sync_modify(j, sourceIndexes)
                                           Return goto_end()
                                       Else
                                           If _dec(times) > 0 Then
                                               Return goto_prev()
                                           Else
                                               raiseError(errorHandle.errorType.warning,
                                                          "failed to sync ", key,
                                                          ", the data may have been deleted already, or lost forever.")
                                               Return goto_end()
                                           End If
                                       End If
                                   End Function))
    End Sub

    Private Sub sync_modify(ByVal i As storageJob,
                            ByVal sourceIndexes As [set](Of UInt32))
        assert(Not i Is Nothing)
        sync_modify(i.key, i.buff, i.i64, sourceIndexes)
    End Sub

    Private Sub sync_modify(ByVal key As String,
                            ByVal buff As bytes,
                            ByVal ts As Int64,
                            ByVal sourceIndexes As [set](Of UInt32))
        assert(valid_sourceIndexes(key, sourceIndexes))
        assert(foreach_redundance_queue(key,
                                        Function(ByRef i As UInt32) As Boolean
                                            If sourceIndexes.find(i) = sourceIndexes.end() Then
                                                Dim j As storageJob = Nothing
                                                j = New storageJob()
                                                j.command = command.modify
                                                j.key = key
                                                j.buff = buff
                                                'force rewrite the timestamp
                                                j.i64 = ts
                                                If j.valid() Then
                                                    assert_begin(New eventComb(ijobStep(j,
                                                                                        targetQueue(i),
                                                                                        AddressOf goto_end)))
                                                Else
                                                    raiseError(errorHandle.errorType.exclamation,
                                                               "cannot create a valid storageJob to sync ", key,
                                                               ", the data will be lost.")
                                                End If
                                            End If
                                            Return True
                                        End Function))
    End Sub
End Class
